iT邦幫忙

2025 iThome 鐵人賽

DAY 8
0

在前面的文章中,我們從非同步概念、協程、事件迴圈,一路探索到執行緒、程序以及 GIL。現在,讓我們用這些知識來回答一個所有 FastAPI 開發者都會遇到的核心問題:

在 FastAPI 中,defasync def 定義的 API 路由,到底有什麼差別?我該在什麼時候用哪一個?

深入原理

要做出最好的選擇,我們必須理解 FastAPI 在背後做了什麼。接下來讓我們串連起之前學到的所有知識。

async def

當你使用 async def 定義路由時,你的程式碼將直接在 FastAPI 底層的事件迴圈 (Event Loop) 上執行(由 Uvicorn 等 ASGI 伺服器提供)。這種執行方式的精髓在於 await 關鍵字:當事件迴圈執行到 await 一個 I/O 操作時(例如 await db.query()),它不會原地等待,而是會「掛起」當前任務,立即轉去處理其他進來的請求。當原本的 I/O 操作完成後,再回來繼續執行剩餘的程式碼。

這種機制讓單一執行緒就能實現極高的並行處理能力 (Concurrency),特別適合處理大量的網路請求。

程式碼範例

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 使用非同步資料庫驅動 (如 databases, asyncpg)
    user_data = await db.users.get(id=user_id) 
    # 使用非同步 HTTP 客戶端 (如 httpx)
    external_data = await httpx.get(f"https://api.example.com/data/{user_id}")
    return {"user": user_data, "external": external_data.json()}

def

相對地,當你使用 def 定義路由時,FastAPI 會採取完全不同的處理策略。由於 FastAPI 知道這個函式可能會包含阻塞操作,為了保護珍貴的事件迴圈不被卡住,它不會讓這些函式直接在主事件迴圈上執行。取而代之的是,FastAPI 會從一個外部執行緒池 (External Thread Pool) 中分配一個執行緒,然後將你的 def 函式放到那個獨立的執行緒中執行。

這樣一來,即使函式內部有 time.sleep(10) 或複雜的 CPU 運算,也只會阻塞那個被分配到的執行緒,而主事件迴圈依然能夠暢通無阻地接收和處理其他請求(特別是 async def 的請求)。

程式碼範例

def complex_cpu_calculation(data: dict) -> int:
    # 這裡是一個耗時的 CPU 運算
    # ...
    time.sleep(10) # 模擬阻塞操作
    return sum(data.values())

@app.post("/calculate")
def run_calculation(data: dict):
    result = complex_cpu_calculation(data)
    return {"result": result}

效能的邊界:GIL 的限制

既然 def 會在執行緒池中執行,這是否表示我們可以透過 def 路由來實現 CPU 密集型任務的並列 (Parallelism),從而利用多核心 CPU 呢?

答案:不行,因為有 GIL。

GIL 會確保同一個程序中,任何時候都只有一個執行緒在執行 Python 位元組碼。即使你的伺服器有 16 個核心,多個執行緒的 def 路由也無法「真正地同時」進行 CPU 運算。它們會爭搶 GIL,導致頻繁的上下文切換,效能甚至可能比單執行緒更差。

決策框架

現在,你可以根據以下簡單的原則來做決定:

使用 async def

  • 當你的 API 含有 I/O 密集型 (I/O-Bound) 操作時
  • 範例:請求外部 API、查詢資料庫、讀寫檔案、等待 WebSocket 訊息
  • 原因:async def 讓 FastAPI 能在「等待」I/O 時處理其他請求,最大化伺服器吞吐量

使用 def

  • 當你的 API 含有 CPU 密集型 (CPU-Bound) 操作,或是你正在使用一個不支援 asyncio 的傳統阻塞函式庫時
  • 範例:處理圖像、影片編碼、複雜的數學運算、機器學習預測
  • 原因:FastAPI 會聰明地將 def 路由放到一個獨立的執行緒池中執行,避免阻塞主事件迴圈

常見陷阱與最佳實踐

最大陷阱:在 async def 中呼叫阻塞函式!

這是最常見也最嚴重的錯誤:

import requests

@app.get("/bad-request")
async def bad_request():
    # requests.get() 是一個阻塞函式
    # 它會凍結整個事件迴圈,所有其他請求都會被卡住!
    response = requests.get("https://example.com") 
    return response.json()

修正方法:要嘛改用 def,要嘛換用非同步函式庫(如 httpx)。

如何處理混合情況?

如果一個 async def 函式必須呼叫一個無法避免的阻塞函式,可以使用 fastapi.concurrency.run_in_threadpool

from fastapi.concurrency import run_in_threadpool

def blocking_io_call():
    # ... 
    pass

@app.get("/mixed")
async def handle_mixed():
    # 將阻塞函式放到執行緒池,然後 await 它的結果
    # 這能避免阻塞事件迴圈
    result = await run_in_threadpool(blocking_io_call)
    return {"status": "ok"}

有關 thread pool 的部分,會在後面的文章做更完整的介紹。

小結

defasync def 在 FastAPI 中並非對立,而是相輔相成的工具,它們共同構成了框架的彈性與強大。

  • async def 是 FastAPI 的「預設」與「未來」:它利用事件迴圈處理 I/O 密集型任務,以實現最大的伺服器效能。
  • def 是強大的「兼容模式」:它利用執行緒池確保了 CPU 密集型任務和傳統阻塞函式庫不會拖垮整個應用。

理解了它們背後的非同步、執行緒與 GIL 原理,你就能夠針對每個具體場景,設計出最高效、最穩定的 FastAPI 應用程式。


上一篇
[Day 07] GIL
下一篇
[Day 09] WSGI vs. ASGI
系列文
用 FastAPI 打造你的 AI 服務22
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言